package com.asiainfo.dacp.datastash.broker.flow.listener.service;

import java.util.List;

import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.common.constant.ZKConstant;
import com.asiainfo.dacp.datastash.broker.common.enums.UnitStateEnum;
import com.asiainfo.dacp.datastash.broker.core.BrokerModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitTaskStateEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.Broker;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.facade.service.LinearProgramming;
import com.asiainfo.dacp.datastash.broker.flow.core.domain.Flow;
import com.asiainfo.dacp.datastash.broker.flow.core.service.FlowStateChangeService;
import com.asiainfo.dacp.datastash.broker.redis.service.RedisFlowService;
import com.asiainfo.dacp.datastash.broker.util.ZKUtil;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;

@Service
public class TaskWaitProcessor {
	private static Logger LOG = LoggerFactory.getLogger(TaskWaitProcessor.class);

	@Autowired
	private RedisFlowService redisService;

	@Autowired
	private FlowStateChangeService stateService;

	@Autowired
	private BrokerModel brokerFacade;
	@Autowired
	private LinearProgramming linearProgramming;

	@Async
	protected void handle() {
		String serNo = redisService.fetchTasksWait();
		if (Strings.isNullOrEmpty(serNo))
			return;
		LOG.info(String.format("[%s] 开始进行 [单元任务] 依赖判断.", serNo));

		String lockPath = String.format(ZKConstant.UNIT_LOCK, serNo.substring(0, serNo.indexOf("-")));
		Flow flow = null;
		InterProcessMutex lock = null;
		try {
			lock = ZKUtil.getLock(lockPath);
			LOG.info(String.format("[%s] 获取分布式锁 [%s].", serNo, lockPath));
			lock.acquire();

			flow = redisService.getFlow(serNo);
			flow.setUnitState(UnitStateEnum.RUNNING);
			List<StashUnitTask> tasks = flow.getReadyTasks();
			LOG.info(String.format("[%s] 满足运行条件的 [单元任务] 数:: %d.", serNo, tasks.size()));

			if (!tasks.isEmpty()) {
				for (StashUnitTask task : tasks) {
					Broker broker = linearProgramming.programming(task, brokerFacade.brokers());

					Preconditions.checkNotNull(broker);
					task.setBrokerId(broker.getBrokerId());
					task.setBrokerId(broker.getBrokerIp());
					task.setTaskState(UnitTaskStateEnum.RUNNABLE);

					LOG.info(String.format("[%s] [单元任务] 满足运行条件，等待 broker[%s] 执行.", task.buildSerNo(),
							broker.getBrokerIp()));
				}
				stateService.LineOK(flow);
				redisService.putTaskWork(tasks);
			}
			redisService.createOrUpdate(flow);
		} catch (Exception e) {
			stateService.LineNotOK(flow);
			LOG.error(String.format("[%s] 单元任务运行条件判断 过程失败，请确认.", serNo), e);
		} finally {
			try {
				if (lock != null)
					lock.release();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

	}

}
